package com.mos.iot.mqtt.callback;

import com.mos.iot.mqtt.MqttClientManager;
import com.mos.iot.mqtt.listener.MethodParam;
import com.mos.iot.mqtt.listener.MqttTopicListenerManager;
import com.mos.iot.mqtt.listener.MqttTopicMethodManager;
import cn.hutool.core.text.AntPathMatcher;
import cn.hutool.core.util.ReflectUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import reactor.core.publisher.Mono;

import java.lang.reflect.Method;
import java.util.List;
import java.util.Map;

@Slf4j
@RequiredArgsConstructor
public class MqttClientRoute implements MqttCallback {

    private final AntPathMatcher pathMatcher = new AntPathMatcher();

    private final String clientId;

    private final MqttConnectOptions connectOptions;

    @Override
    public void connectionLost(Throwable cause) {
        try {
            if (null != clientId) {
                if (null != connectOptions) {
                    MqttClientManager.getMqttClient(clientId).connect(connectOptions);
                } else {
                    MqttClientManager.getMqttClient(clientId).connect();
                }
            }

        } catch (Exception e) {
            log.error("{} reconnect failed!", clientId, e);
        }
    }

    /**
     * 接收订阅消息
     * @param topic 主题
     * @param message 接收消息
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        Map<String, List<Map<Method, Object>>> listener = MqttTopicListenerManager.getListener(this.clientId);
        if(listener != null && !listener.isEmpty()) {
            for (Map.Entry<String, List<Map<Method, Object>>> entry : listener.entrySet()) {
                String subscriptionTopic = entry.getKey();
                List<Map<Method, Object>> methods = entry.getValue();
                if(pathMatcher.match(subscriptionTopic, topic)) {
                    Map<String, String> topicParams = pathMatcher.extractUriTemplateVariables(subscriptionTopic, topic);
                    for (Map<Method, Object> method : methods) {
                        for (Map.Entry<Method, Object> methodObjectEntry : method.entrySet()) {
                            Method targetMethod = methodObjectEntry.getKey();
                            MethodParam methodParam = MqttTopicMethodManager.get(targetMethod);
                            Mono.create(sink -> {
                                Object[] params = methodParam.getParams(topic, topicParams, message);
                                Object bean = methodObjectEntry.getValue();
                                sink.success(ReflectUtil.invoke(bean, targetMethod, params));
                            }).subscribe(res -> {
                                if(res != null) {
                                    MqttClientManager.publish(methodParam.getOutMessage(res, topicParams));
                                }
                            });
                        }
                    }
                }
            }
        }
    }

    /**
     * 消息发送成功
     * @param token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

    }
}
